package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 个股分时窗口函数
 */
public class StockMinWindowFunction extends RichWindowFunction<CleanBean, StockBean, String, TimeWindow> {


    /**
     * 开发步骤：
     * 1.初始化 MapState<String, StockBean>
     * 2.记录最新个股
     * 3.获取分时成交额和成交数量
     * 4.格式化日期
     * 5.封装输出数据
     * 6.更新MapState
     */
    //1.初始化 MapState<String, StockBean>
    MapState<String, StockBean> msState = null;
    @Override
    public void open(Configuration parameters) throws Exception {

        msState = getRuntimeContext().getMapState(new MapStateDescriptor<String, StockBean>("msState", String.class, StockBean.class));

    }

    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<StockBean> out) throws Exception {

        //2.记录最新个股
        CleanBean line = null;
        for (CleanBean cleanBean : input) {
            if(line == null){
                line = cleanBean;
            }
            if(line.getEventTime() < cleanBean.getEventTime()){
                line = cleanBean;
            }
        }

        //3.获取分时成交额和成交数量
        //3.1获取上一分钟的总成交量和总成交金额
        StockBean stockBeanLast = msState.get(line.getSecCode());
        long minAmt = 0l;
        long minVol = 0l;
        if(stockBeanLast != null){
            //3.2 计算分时成交量和金额
             minAmt = line.getTradeAmt() - stockBeanLast.getTradeAmtDay();
             minVol = line.getTradeVolumn() - stockBeanLast.getTradeVolDay();
        }

        //4.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        String tradeDate = DateUtil.longTimeToString(line.getEventTime(), Constant.format_yyyy_mm_dd);

        //5.封装输出数据
        StockBean stockBean = new StockBean(
                tradeTime,
                line.getSecCode(),
                line.getSecName(),
                line.getPreClosePrice(),
                line.getOpenPrice(),
                line.getHighPrice(),
                line.getLowPrice(),
                line.getTradePrice(),
                minVol, minAmt,
                line.getTradeVolumn(),
                line.getTradeAmt(),
                tradeDate,
                line.getEventTime(),
                line.getSource()
        );
        out.collect(stockBean);

        //6.更新MapState:把最新的个股分时数据缓存进去
        msState.put(stockBean.getSecCode(),stockBean);

    }
}
